package com.kai.mqtt.protocol;

import com.kai.mqtt.auth.service.IMqttAclService;
import com.kai.mqtt.constant.AttributeKeyConstant;
import com.kai.mqtt.enums.AclActionEnum;
import com.kai.mqtt.message.DupPublishMessage;
import com.kai.mqtt.message.RetainMessage;
import com.kai.mqtt.message.SubscribeMessage;
import com.kai.mqtt.message.service.IDupPublishMessageService;
import com.kai.mqtt.message.service.IMessageIdService;
import com.kai.mqtt.message.service.IRetainMessageService;
import com.kai.mqtt.message.service.ISubscribeMessageService;
import com.kai.mqtt.session.Session;
import com.kai.mqtt.session.service.ISessionService;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;

/**
 * @author weijk
 * @since 2024/6/3
 */
@Slf4j
@Component
public class Publish extends AbstractProtocol {

    @Resource
    private IRetainMessageService retainMessageService;

    @Resource
    private ISubscribeMessageService subscribeMessageService;

    @Resource
    private ISessionService sessionService;

    @Resource
    private IMessageIdService messageIdService;

    @Resource
    private IDupPublishMessageService dupPublishMessageService;

    @Autowired
    private IMqttAclService mqttAclService;

    @Override
    public void process(Channel channel, MqttMessage message) {
        MqttPublishMessage msg = (MqttPublishMessage) message;

        Session session = sessionService.get(channel.attr(AttributeKeyConstant.CLIENT_ID).get());
        String topicName = msg.variableHeader().topicName();
        if (!mqttAclService.checkValid(session.getUsername(), session.getClientId(), AclActionEnum.PUBLISH, topicName)) {
            return;
        }

        byte[] messageBytes = new byte[msg.payload().readableBytes()];
        msg.payload().getBytes(msg.payload().readerIndex(), messageBytes);
        log.info("receive - publish, clientId: {}, topic: {}, qos: {}, content: {}",
                session.getClientId(), topicName, msg.fixedHeader().qosLevel().value(), new String(messageBytes));
        if (Objects.equals(msg.fixedHeader().qosLevel(), MqttQoS.AT_MOST_ONCE)) {
            sendPublishMessage(topicName, msg.fixedHeader().qosLevel(), messageBytes);
        }
        if (Objects.equals(msg.fixedHeader().qosLevel(), MqttQoS.AT_LEAST_ONCE)) {
            sendPublishMessage(topicName, msg.fixedHeader().qosLevel(), messageBytes);
            sendPubAckMessage(channel, msg.variableHeader().packetId());
        }
        if (Objects.equals(msg.fixedHeader().qosLevel(), MqttQoS.EXACTLY_ONCE)) {
            sendPublishMessage(topicName, msg.fixedHeader().qosLevel(), messageBytes);
            sendPubRecMessage(channel, msg.variableHeader().packetId());
        }
        if (msg.fixedHeader().isRetain()) {
            if (Objects.equals(messageBytes.length, 0)) {
                retainMessageService.remove(topicName);
            } else {
                RetainMessage retainMessage = new RetainMessage()
                        .setTopic(topicName)
                        .setQos(msg.fixedHeader().qosLevel().value())
                        .setClientId(session.getClientId())
                        .setMessageBytes(messageBytes);
                retainMessageService.put(topicName, retainMessage);
            }
        }
    }

    private void sendPublishMessage(String topic, MqttQoS mqttQoS, byte[] messageBytes) {
        List<SubscribeMessage> subscribeMessages = subscribeMessageService.search(topic);
        subscribeMessages.forEach(subscribeMessage -> {
            String clientId = subscribeMessage.getClientId();
            MqttQoS respQoS = mqttQoS.value() > subscribeMessage.getQos() ? MqttQoS.valueOf(subscribeMessage.getQos()) : mqttQoS;
            if (Objects.equals(respQoS, MqttQoS.AT_MOST_ONCE)) {
                MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, false, 0),
                        new MqttPublishVariableHeader(topic, 0), Unpooled.buffer().writeBytes(messageBytes));
                log.debug("send - publish, clientId: {}, topic: {}, qos: {}, content: {}", clientId, topic, respQoS.value(), new String(messageBytes));
                sessionService.get(clientId).getChannel().writeAndFlush(publishMessage);
            }
            if (Objects.equals(respQoS, MqttQoS.AT_LEAST_ONCE) || Objects.equals(respQoS, MqttQoS.EXACTLY_ONCE)) {
                Integer messageId = messageIdService.get(clientId);
                MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, false, 0),
                        new MqttPublishVariableHeader(topic, messageId), Unpooled.buffer().writeBytes(messageBytes));
                log.debug("send - publish, clientId: {}, topic: {}, qos: {}, messageId: {}, content: {}", clientId, topic, respQoS.value(), messageId, new String(messageBytes));
                DupPublishMessage dupPublishMessage = new DupPublishMessage()
                        .setTopic(topic)
                        .setQos(respQoS.value())
                        .setMessageId(messageId)
                        .setMessageBytes(messageBytes);
                dupPublishMessageService.put(clientId, dupPublishMessage);
                sessionService.get(clientId).getChannel().writeAndFlush(publishMessage);
            }
        });
    }

    private void sendPubAckMessage(Channel channel, Integer messageId) {
        MqttPubAckMessage pubAckMessage = (MqttPubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(messageId), null);
        log.debug("send - pubAck, clientId: {}, messageId: {}", channel.attr(AttributeKeyConstant.CLIENT_ID).get(), messageId);
        channel.writeAndFlush(pubAckMessage);
    }

    private void sendPubRecMessage(Channel channel, Integer messageId) {
        MqttMessage pubRecMessage = MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(messageId), null);
        log.debug("send - pubRec, clientId: {}, messageId: {}", channel.attr(AttributeKeyConstant.CLIENT_ID).get(), messageId);
        channel.writeAndFlush(pubRecMessage);
    }
}
